package func.udf;

import api.beans.SensorReading;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;

import java.util.List;

public class UDFTest4 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        DataStream<String> inputStream = env.readTextFile("D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor.txt");
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0],new Long(fields[1]), new Double(fields[2]));
        });
        Table sensorTable = tableEnv.fromDataStream(dataStream);
        String udfPath = "func.model.";
        String listStr = "[\"HashCode\"]";

        List<String> list = JSONObject.parseArray(listStr, String.class);
        for (String udf : list) {
            System.out.println(udf);
            tableEnv.registerFunction(udf, (ScalarFunction) Class.forName(udfPath + udf).newInstance());
        }

        tableEnv.createTemporaryView("sensor", sensorTable);
        Table resultSqlTable = tableEnv.sqlQuery("select id, temperature , HashCode(id) from sensor");
//        Table resultSqlTable = tableEnv.sqlQuery("select id, temperature from sensor");
        // 打印输出
        tableEnv.toAppendStream(resultSqlTable, Row.class).print();

        env.execute();
    }

}
